AlteryxのPython SDKを使ったツールを作る

「カスタマーストーリー」を支えるツール特集

#Python

#Alteryx SDK

#ビッグデータ

#Alteryx

じょんすみす

2018.02.21

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、小澤です。

今回は、Alteryx SDKを使って、Pythonで記述した簡単なツールの作成をしていきます。

作成するツール

今回は簡単なサンプルツールを作成します。処理内容としては、指定した列に指定した値を足すだけというものになります。実際の動きを以下のようなワークフローで確認します。

最初のText Inputツールでは以下のようなデータを作成しています。

続いての"たこ焼き"アイコンのツールが今回作成するツールになります。ツールの設定は以下のようになっています。

対象となる列とたす値を指定します。上記のように、Text Inputツールで入力したnum列に1をたす設定をした場合は以下のような結果が出力されます。

なお、ツールのアイコンにはいらすとやのたこ焼き画像を使用していますが、特に意味はありません。

ツールの実装する

どのようなツールを作るかを確認したので、実際にこのツールのコードを見ていきましょう。

Configファイル

まずは、Configファイルの内容を見ていきます。

<?xml version="1.0"?>
<AlteryxJavaScriptPlugin>
  <EngineSettings EngineDllEntryPoint="python_sdk_test.py" EngineDll="Python" SDKVersion="10.1"/>
  <Properties>
    <MetaInfo>
      <Name>Python SDK Test</Name>
      <Description><![CDATA[This is python sdk test macro.]]></Description>
      <ToolVersion>1</ToolVersion>
      <CategoryName>JS Macro</CategoryName>
      <Author>John Smith</Author>
       <Icon>MacroTest.png</Icon>
    </MetaInfo>
  </Properties>
  <GuiSettings Help="" Html="PythonSDKTestGui.html" Icon="PythonSDKTest.png" SDKVersion="10.1">
    <InputConnections>
      <Connection Name="Input" AllowMultiple="False" Optional="False" Type="Connection"/>
    </InputConnections>
    <OutputConnections>
      <Connection Name="Output" AllowMultiple="False" Optional="False" Type="Connection"/>
    </OutputConnections>
  </GuiSettings>
</AlteryxJavaScriptPlugin>

内容としては、マクロのインターフェースを作った際とほぼ違いはありません。 EngineSettingsのEngineDllにPythonを指定し、EngineDllEnteryPointにPythonのファイルを指定する以外は、これまでで解説したConfigファイルの書き方と同様となっています。

HTML GUI

続いて、HTML GUIの内容を見ていきます。

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>Location tool</title>
    <script type="text/javascript">
      document.write('<link rel="import" href="' + window.Alteryx.LibDir + '2/lib/includes.html">')
    </script>
    <script type="text/javascript">
      window.TestGUINamespace = {};
	    window.TestGUINamespace.item_ui = {
        type: 'DropDown',
        widgetId: 'item'
      }
    </script>
</head>
<body>
  <br />
  <h2>XMSG("Select Field")</h2>
  <ayx data-ui-props='window.TestGUINamespace.item_ui'></ayx>
  <h2>XMSG("Add Value")</h2>
  <ayx data-ui-props='{type: "TextBox", widgetId: "num"}' data-item-props='{dataName: "num", dataType: "SimpleInt"}'></ayx>

  <script type="text/javascript">
    Alteryx.Gui.BeforeLoad = function(manager, AlteryxDataItems, json) {
      var item = new AlteryxDataItems.FieldSelector("item", {manager: manager});
      manager.addDataItem(item);
      manager.bindDataItemToWidget(item, "item");
    };

    window.Alteryx.Gui.Annotation = function (manager) {
      var item = manager.getDataItem('item');
      var num = manager.getDataItem('num');
      return item.getValue() + " -> " + num.getValue();
    };
  </script>
</body>
</html>

こちらも基本的な書き方は同様となります。先ほどの画像のようなインターフェースにするため、ayxタグでDropDownとTextBoxを配置しています。

Pythonのコード

では、いよいよPythonのコードを見ていきます。

インポート

まずは、AlteryxのSDKを利用するのと、設定値を読み込むためのXMLパースライブラリをインポートします。

import AlteryxPythonSDK as Sdk
import xml.etree.ElementTree as Et

AyxPluginクラス

続いて、AyxPluginクラスを見て見ます。

class AyxPlugin:
    def __init__(self, n_tool_id: int, alteryx_engine: object, output_anchor_mgr: object):
        """
        ツールがワークフローに配置されたタイミングで呼び出される
        このメソッドでは処理を行う際に必要な変数の定義などをしておく
        """
        
        # 引数の値をメンバ変数に入れる
        # ほぼ定型となる処理
        self.n_tool_id = n_tool_id
        self.alteryx_engine = alteryx_engine
        self.output_anchor_mgr = output_anchor_mgr

        # プログラム中で利用する設定値を入れておく変数を用意する
        self.item = None
        self.num = None
        
        # 入出力コネクションを入れておく変数を用意する
        self.output_anchor = None
        self.input = None

    def pi_init(self, str_xml: str):
        """
        ツールの設定値が取得可能になったタイミング(ツールの選択が解除されたタイミング)で呼び出される
        XMLで与えられる設定に関する情報から設定値を取り出すのに利用
        """
        
        # 設定値のXMLから必要な値を取り出す
        # 今回の設定項目は計算対象の列を入れる"item"とたす値を入れる"num"の2つなのでそれらを取り出している
        self.item = Et.fromstring(str_xml).find('item').text if 'item' in str_xml else None
        self.num = int(Et.fromstring(str_xml).find('num').text) if 'num' in str_xml else None
        
        # Configファイルで設定された名前を指定して出力先に関する情報をOutputAnchorMessangerから取得する
        # 複数の出力があるようなツールを作る場合はNameで指定した値ごとに個別に取得
        # 実際の処理ではここで設定したoutput_anchorにデータを流し込むことで出力を行う
        self.output_anchor = self.output_anchor_mgr.get_output_anchor('Output')

    def pi_add_incoming_connection(self, str_type: str, str_name: str) -> object:
        """
        入力コネクションが接続された時に呼び出される
        入力に関する情報の設定として、以下の2つを行う
        
        - pre_sortの呼び出し
          -　処理を実行する前にソートをしておく必要があるツールの場合どのような基準でソートするかの設定情報を渡す
        - IncomingInterfaceの生成
          - この後記述するIncomingInterfaceのインスタンスを生成する
        """
        
        # IncomingInterfaceのインスタンスを生成してそれを返す
        # IncomingInterface内の各メソッドも必要なタイミングで呼び出されるのでAlteryx側に渡しておく必要がある
        self.input = IncomingInterface(self)
        return self.input

    def pi_add_outgoing_connection(self, str_name: str) -> bool:
        """
        出力のコネクションが接続された時に呼び出される
        """
        
        # 今回は何もしない
        return True

    def pi_push_all_records(self, n_record_limit: int) -> bool:
        """
        入力コネクションが接続されていない時に呼び出される
        
        入力が必須のツールの場合は、エラー出力を行う
        入力コネクションが存在しないかオプションの場合は、接続されていない時用のデータをここで生成する
        """
        
        # 今回は必須なので、エラーメッセージを出力する
        # AlteryxEngineクラスのoutput_messageメソッドを呼び出すことによって、Resultsのログにメッセージを出力できる
        self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.error, 'Missing Incoming Connection')
        return False

    def pi_close(self, b_has_errors: bool):
        """
        終了処理を行う
        """
        
        # この部分は定型
        self.output_anchor.assert_close()

各処理内容については、docstringおよびコメントを参照してください。

IncomingInterface

IncomingInterface側のコードは以下のようになっています。

class IncomingInterface:
    def __init__(self, parent: object):
        """
        AyxPluginのpi_add_incoming_connectionで生成される
        parentにはAyxPluginのインスタンスが入っているので、必要な情報を取得するのに利用する
        """
        
        self.parent = parent

        # この後の処理で必要な変数を定義しておく
        self.record_info_out = None
        self.record_copier = None

    def ii_init(self, record_info_in: object) -> bool:
        """
        入力コネクションの情報が変化した時に初期設定を行うために呼び出される
        入力されるデータのメタデータが渡されるのでそれを利用して必要な初期化を行う
        """
        
        # 出力のメタデータを生成する
        # 今回は、入力データに1列追加する処理なので、それ以外の列のメタデータは入力のものをコピーする
        self.record_info_out = record_info_in.clone()
        # 追加する列に関するメタデータを設定
        self.record_info_out.add_field("add_num", Sdk.FieldType.int32)

        # 実際に処理を行う際に各出力レコードへ入力のデータをそのままコピーするために必要な設定を行う
        self.record_copier = Sdk.RecordCopier(self.record_info_out, record_info_in)
        for index in range(record_info_in.num_fields):
            self.record_copier.add(index, index)
        self.record_copier.done_adding()

        # 出力のコネクションにこのツールが出力するメタデータを渡す
        self.parent.output_anchor.init(self.record_info_out)
        return True

    def ii_push_record(self, in_record: object) -> bool:
        """
        ワークフローが実行された際に呼び出されるメソッド
        データのうち1レコードが渡される
        そのため、このメソッドはデータの行数と同じ回数呼び出される
        """
        
        # 出力のメタデータの内容に基づいて出力のレコードを生成する
        record_creator = self.record_info_out.construct_record_creator()

        # ii_initで設定した対応関係に基づいて、入力のデータを出力にコピーする
        self.record_copier.copy(record_creator, in_record)
        
        # 計算対象の列を指定する
        input_field = self.record_info_out[self.record_info_out.get_field_num(self.parent.item)]
        
        # 出力対象となる列を指定する
        output_field = self.record_info_out[self.record_info_out.get_field_num("add_num")]

        # 計算対象の列のデータを取得する
        num = input_field.get_as_int32(in_record)
        
        # 計算を行なって、その結果を出力対象の列に設定する
        output_field.set_from_int32(record_creator, num+self.parent.num)

        out_record = record_creator.finalize_record()

        # データを出力する
        if not self.parent.output_anchor.push_record(out_record):
            return False
        return True

    def ii_update_progress(self, d_percent: float):
        """
        進捗に関する情報を渡す
        特にこだわりがなければ定型で問題ない
        """
        
        self.parent.alteryx_engine.output_tool_progress(self.parent.n_tool_id, d_percent)
        self.parent.output_anchor.update_progress(d_percent)

    def ii_close(self):
        """
        終了処理を行う
        """
        
        # この部分は定型
        self.parent.output_anchor.close()

こちらもdocstringおよびコメントにて処理内容を解説しています。

RecordとField

さて、この一連の処理の中で簡単にちょっと複雑な対応関係になっているのが、Record関連とFieldクラスを使ったデータの取得や生成部分になります。これらの関係を整理しておくことで、ii_push_recordないの処理内容などがスッキリするかと思いますので、ここではその関係を解説していきます。

各クラスの対応関係は以下のようになっています。

クラス	役割
RecordInfo	各列のメタデータ
RecordRef	実際のデータ
Field	RecordRef中の個々の列の値を取得したり設定したりする
RecordCreator	RecordRefを作成する
RecordCopier	RecordRefの中の値を既存のものからコピーする

という対応関係になっています。 Alteryxのデータのと対応関係としては、以下のようになってるとイメージすればわかりやすいかと思います。

Alteryxでは同一列のデータは全て列名や型などの性質が同じで、かつ途中で列が増えたり減ったりすることはありません。そのため、全てに共通する属性としてRecordInfoがまず一番大元の情報となります。他のクラスは、全てこのRecordInfoに基づいて生成されます。

この対応関係の確認のために、改めて先ほどのコードを見てみます。

record_creator = self.record_info_out.construct_record_creator()
...
out_record = record_creator.finalize_record()

まず、RecordCreatorはRecordInfoの情報に基づいて生成されています。その上で、作成するデータに関する値の設定を行なったのち、finalize_recordで出力となるRecordRefを生成しています。

input_field = self.record_info_out[self.record_info_out.get_field_num(self.parent.item)]
...
num = input_field.get_as_int32(in_record)

FieldもRecordInfoの列名(列番号)を指定してメタデータ中の列の情報をまず取得しています。その上で、実際にデータの取得や設定を行う際には対応するRecordRefをFieldクラスのメソッドに渡すことで値を取得しています。

self.record_copier = Sdk.RecordCopier(self.record_info_out, record_info_in)
for index in range(record_info_in.num_fields):
    self.record_copier.add(index, index)
self.record_copier.done_adding()
...
self.record_copier.copy(record_creator, in_record)

RecordCopierでは、まず参照元と参照先の2つのRecordInfo間での各列の対応関係を定義しています。その上でコピー先となるRecordCreatorとコピー元であるRecordRefを引数にとって値のコピーを行なっています。

このようにAlteryx SDKでのデータ操作はRecordInfoを全ての大元としてその情報に基づいた各種操作を行うためのクラスを生成し、それを経由して実際の値の設定を行うという流れになります。最終的に必要な情報が入ったRecordRefはOutputAnchorに流してやればOKです。